package com.geek.framework.study.event.zookeeper;

import com.geek.framework.study.event.EventListener;
import com.geek.framework.study.event.distributed.DistributedEventObject;
import com.geek.framework.study.event.reactive.stream.ListenerSubscriberAdapter;
import com.geek.framework.study.reactive.quickstart.SimplePublisher;

import java.io.Serializable;
import java.util.EventObject;

/**
 * @Description 分布消息队列发布器抽象类
 */
public abstract class AbstractDistributedEventPublisher implements AutoCloseable {
    protected final SimplePublisher<EventObject> simplePublisher;

    public AbstractDistributedEventPublisher() {
        this.simplePublisher = new SimplePublisher();
        addDistributedEventListener();
    }

    public void publish(Serializable event) {
        simplePublisher.publish(new DistributedEventObject(event));
    }

    public void addEventListener(EventListener eventListener) {
        simplePublisher.subscribe(new ListenerSubscriberAdapter(eventObject -> {
            if (!(eventObject instanceof DistributedEventObject)) {
                eventListener.onEvent(eventObject);
            }
        }));
        afterAddEventListener(eventListener);
    }

    /**
     * 添加完本地监听器后的回调，可以做一些监听消息引擎队列的操作
     * @param eventListener
     */
    protected abstract void afterAddEventListener(EventListener eventListener);

    /**
     * 判断是否是需要分发的业务事件
     */
    private void addDistributedEventListener() {
        simplePublisher.subscribe(new ListenerSubscriberAdapter(event -> {
            if (event instanceof DistributedEventObject) {
                try {
                    pushToMessagingEngine(event.getSource());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }));
    }

    /**
     * 发布到消息引擎去
     *
     * @param event 该数据为方法 {@link AbstractDistributedEventPublisher#publish(Serializable)}传入的数据
     * @throws Exception
     */
    protected abstract void pushToMessagingEngine(Object event) throws Exception;

    /**
     * 从消息引擎接收
     *
     * @param event 该数据为方法 {@link AbstractDistributedEventPublisher#publish(Serializable)}传入的数据
     */
    protected void receiveFromMessagingEngine(Object event) {
        simplePublisher.publish(new EventObject(event));
    }

    @Override
    public void close() {
        doClose();
    }

    /**
     * 可以在该方法做一些清理，如释放消息引擎的连接等等
     */
    protected void doClose() {
    }
}
